package com.ztorn.cdc;

import com.alibaba.fastjson.JSONObject;
import com.ztorn.cdc.doris.DorisSinkOptions;
import com.ztorn.common.assertion.Asserts;
import com.ztorn.common.model.*;
import com.ztorn.common.util.JSONUtil;
import com.ztorn.common.util.SqlUtil;
import com.ztorn.driver.DriverSchema;
import com.ztorn.framework.driver.Driver;
import lombok.extern.slf4j.Slf4j;
import org.apache.doris.flink.cfg.DorisExecutionOptions;
import org.apache.doris.flink.cfg.DorisOptions;
import org.apache.doris.flink.cfg.DorisReadOptions;
import org.apache.doris.flink.sink.DorisSink;
import org.apache.doris.flink.sink.writer.JsonDebeziumSchemaSerializer;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.table.data.*;
import org.apache.flink.table.operations.ModifyOperation;
import org.apache.flink.table.types.logical.*;
import org.apache.flink.types.RowKind;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.math.BigDecimal;
import java.sql.Timestamp;
import java.time.Instant;
import java.time.ZoneId;
import java.util.*;

@Slf4j
public abstract class AbstractSinkBuilder implements SinkBuilder{

    public static final String METADATA_TYPE = "MySql";

    protected FlinkCDCConfig config;
    protected List<ModifyOperation> modifyOperations = new ArrayList();

    public AbstractSinkBuilder() {}

    public AbstractSinkBuilder(FlinkCDCConfig config) {
        this.config = config;
    }

    public FlinkCDCConfig getConfig() {
        return config;
    }

    public void setConfig(FlinkCDCConfig config) {
        this.config = config;
    }

    protected Properties getProperties() {
        Properties properties = new Properties();
        Map<String, String> sink = config.getSink();
        for (Map.Entry<String, String> entry : sink.entrySet()) {
            if (Asserts.isNotNullString(entry.getKey())
                    && entry.getKey().startsWith("properties")
                    && Asserts.isNotNullString(entry.getValue())) {
                properties.setProperty(entry.getKey().replace("properties.", ""), entry.getValue());
            }
        }
        return properties;
    }

    protected SingleOutputStreamOperator<Map> deserialize(
            DataStreamSource<String> dataStreamSource) {
        return dataStreamSource.map(
                new MapFunction<String, Map>() {
                    @Override
                    public Map map(String value) throws Exception {
                        ObjectMapper objectMapper = new ObjectMapper();
                        return objectMapper.readValue(value, Map.class);
                    }
                });
    }

    private boolean executeDDL(Map value,String schema,Map<String, String> sink) {
        try {
            String historyRecord = (String)value.get("historyRecord");
            JSONObject hisObj = JSONObject.parseObject(historyRecord);
            Map<String,String> hisMap = new HashMap<>();
            Iterator<String> hisIterator = hisObj.keySet().iterator();
            while(hisIterator.hasNext())
            {
                String k = hisIterator.next();
                String v = hisObj.getString(k);
                hisMap.put(k, v);
            }
            String ddl = hisMap.get("ddl");
            Driver driver = DriverSchema.checkAndCreateSinkSchema(sink, schema);
            driver.execute(ddl);
            return true;
        }catch (Exception e){
            log.error("table structure is change,execute ddl error,{}",e.getMessage());
            return false;
        }
    }

    protected SingleOutputStreamOperator<Map> shunt(
            SingleOutputStreamOperator<Map> mapOperator, Table table, String schemaFieldName,CDCBuilder cdcBuilder) {
        final String tableName = table.getName();
        final String schemaName = table.getSchema();
        Map<String, String> sink = config.getSink();
        return mapOperator.filter(
                new FilterFunction<Map>() {
                    @Override
                    public boolean filter(Map value) throws Exception {
                        LinkedHashMap source = (LinkedHashMap) value.get("source");
//                        if(value.containsKey("historyRecord")) {
//                            executeDDL(value, schemaName, sink);
//                        }
                        return tableName.equals(source.get("table").toString())
                                && schemaName.equals(source.get(schemaFieldName).toString());
                    }
                });
    }

    protected DataStream<Map> shunt(
            SingleOutputStreamOperator<Map> processOperator, Table table, OutputTag<Map> tag) {

        return processOperator.getSideOutput(tag);
    }

    protected DataStream<RowData> buildRowData(
            SingleOutputStreamOperator<Map> filterOperator,
            List<String> columnNameList,
            List<LogicalType> columnTypeList,
            Table table) {

        return filterOperator.flatMap(
                new FlatMapFunction<Map, RowData>() {
                    @Override
                    public void flatMap(Map value, Collector<RowData> out) throws Exception {
                        try {
                            if(value.containsKey("op")) {
                                switch (value.get("op").toString()) {
                                    case "r":
                                    case "c":
                                        GenericRowData igenericRowData =
                                                new GenericRowData(columnNameList.size());
                                        igenericRowData.setRowKind(RowKind.INSERT);
                                        Map idata = (Map) value.get("after");
                                        for (int i = 0; i < columnNameList.size(); i++) {
                                            igenericRowData.setField(
                                                    i,
                                                    convertValue(
                                                            idata.get(columnNameList.get(i)),
                                                            columnTypeList.get(i)));
                                        }
                                        out.collect(igenericRowData);
                                        break;
                                    case "d":
                                        GenericRowData dgenericRowData =
                                                new GenericRowData(columnNameList.size());
                                        dgenericRowData.setRowKind(RowKind.DELETE);
                                        Map ddata = (Map) value.get("before");
                                        for (int i = 0; i < columnNameList.size(); i++) {
                                            dgenericRowData.setField(
                                                    i,
                                                    convertValue(
                                                            ddata.get(columnNameList.get(i)),
                                                            columnTypeList.get(i)));
                                        }
                                        out.collect(dgenericRowData);
                                        break;
                                    case "u":
                                        GenericRowData ubgenericRowData =
                                                new GenericRowData(columnNameList.size());
                                        ubgenericRowData.setRowKind(RowKind.UPDATE_BEFORE);
                                        Map ubdata = (Map) value.get("before");
                                        for (int i = 0; i < columnNameList.size(); i++) {
                                            ubgenericRowData.setField(
                                                    i,
                                                    convertValue(
                                                            ubdata.get(columnNameList.get(i)),
                                                            columnTypeList.get(i)));
                                        }
                                        out.collect(ubgenericRowData);
                                        GenericRowData uagenericRowData =
                                                new GenericRowData(columnNameList.size());
                                        uagenericRowData.setRowKind(RowKind.UPDATE_AFTER);
                                        Map uadata = (Map) value.get("after");
                                        for (int i = 0; i < columnNameList.size(); i++) {
                                            uagenericRowData.setField(
                                                    i,
                                                    convertValue(
                                                            uadata.get(columnNameList.get(i)),
                                                            columnTypeList.get(i)));
                                        }
                                        out.collect(uagenericRowData);
                                        break;
                                    default:
                                }
                            }
                        } catch (Exception e) {
                            log.error(
                                    "SchameTable: {} - Row: {} - Exception:",
                                    table.getSchemaTableName(),
                                    JSONUtil.toJsonString(value),
                                    e);
                            throw e;
                        }
                    }
                });
    }

    public abstract void addSink(
            StreamExecutionEnvironment env,
            DataStream<RowData> rowDataDataStream,
            Table table,
            List<String> columnNameList,
            List<LogicalType> columnTypeList);

//    @Override
//    public DataStreamSource build(
//            CDCBuilder cdcBuilder,
//            StreamExecutionEnvironment env,
//            DataStreamSource<String> dataStreamSource) {
//
//        final List<Schema> schemaList = config.getSchemaList();
//        final String schemaFieldName = config.getSchemaFieldName();
//        if (Asserts.isNotNullCollection(schemaList)) {
//            SingleOutputStreamOperator<Map> mapOperator = deserialize(dataStreamSource);
//            for (Schema schema : schemaList) {
//                for (Table table : schema.getTables()) {
//                    List<String> columnNameList = new ArrayList<>();
//                    List<LogicalType> columnTypeList = new ArrayList<>();
//                    buildColumn(columnNameList, columnTypeList, table.getColumns());
//                    SingleOutputStreamOperator<Map> filterOperator = shunt(mapOperator,table,schemaFieldName,cdcBuilder);
//
//                    DataStream<RowData> rowDataDataStream =
//                            buildRowData(
//                                    filterOperator,
//                                    columnNameList,
//                                    columnTypeList,
//                                    table);
//
//                    addSink(env, rowDataDataStream, table, columnNameList, columnTypeList);
//                }
//            }
//        }
//        return dataStreamSource;
//    }

    @Override
    public DataStreamSource build(
            CDCBuilder cdcBuilder,
            StreamExecutionEnvironment env,
            DataStreamSource<String> dataStreamSource) {

        List<String> excludeTableList = config.getExcludeTableList();
        Map<String, String> sink = config.getSink();

        Properties properties = getProperties();
        // schema evolution need json format
        properties.setProperty("format", "json");
        properties.setProperty("read_json_by_line", "true");

        Map<Table, OutputTag<String>> tagMap = new HashMap<>();
        Map<String, Table> tableMap = new HashMap<>();
        ObjectMapper objectMapper = new ObjectMapper();

        SingleOutputStreamOperator<Map> mapOperator =
                dataStreamSource.map(x -> objectMapper.readValue(x, Map.class)).returns(Map.class);
        final List<Schema> schemaList = config.getSchemaList();
        final String schemaFieldName = config.getSchemaFieldName();
        if (Asserts.isNotNullCollection(schemaList)) {
            for (Schema schema : schemaList) {
                for (Table table : schema.getTables()) {
                    String sinkTableName = getSinkTableName(table);
                    OutputTag<String> outputTag = new OutputTag<String>(sinkTableName) {};
                    tagMap.put(table, outputTag);
                    tableMap.put(table.getSchemaTableName(), table);
                }
            }
            SingleOutputStreamOperator<String> process =
                    mapOperator.process(
                            new ProcessFunction<Map, String>() {

                                @Override
                                public void processElement(
                                        Map map, Context ctx, Collector<String> out)
                                        throws Exception {
                                    LinkedHashMap source = (LinkedHashMap) map.get("source");
                                    try {
                                        String result = objectMapper.writeValueAsString(map);
                                        Table table =
                                                tableMap.get(
                                                        source.get(schemaFieldName).toString()
                                                                + "."
                                                                + source.get("table").toString());
                                        OutputTag<String> outputTag = tagMap.get(table);
                                        ctx.output(outputTag, result);
                                    } catch (Exception e) {
                                        out.collect(objectMapper.writeValueAsString(map));
                                    }
                                }
                            });
            tagMap.forEach(
                    (table, v) -> {
                        DorisOptions dorisOptions =
                                DorisOptions.builder()
                                        .setFenodes(
                                                config.getSink()
                                                        .get(DorisSinkOptions.FENODES.key()))
                                        .setTableIdentifier(
                                                getSinkSchemaName(table)
                                                        + "."
                                                        + getSinkTableName(table))
                                        .setUsername(
                                                config.getSink()
                                                        .get(DorisSinkOptions.USERNAME.key()))
                                        .setPassword(
                                                config.getSink()
                                                        .get(DorisSinkOptions.PASSWORD.key()))
                                        .build();

                        DorisExecutionOptions.Builder executionBuilder =
                                DorisExecutionOptions.builder();
                        if (sink.containsKey(DorisSinkOptions.SINK_BUFFER_COUNT.key())) {
                            executionBuilder.setBufferCount(
                                    Integer.valueOf(
                                            sink.get(DorisSinkOptions.SINK_BUFFER_COUNT.key())));
                        }
                        if (sink.containsKey(DorisSinkOptions.SINK_BUFFER_SIZE.key())) {
                            executionBuilder.setBufferSize(
                                    Integer.valueOf(
                                            sink.get(DorisSinkOptions.SINK_BUFFER_SIZE.key())));
                        }
                        if (sink.containsKey(DorisSinkOptions.SINK_ENABLE_DELETE.key())) {
                            executionBuilder.setDeletable(
                                    Boolean.valueOf(
                                            sink.get(DorisSinkOptions.SINK_ENABLE_DELETE.key())));
                        } else {
                            executionBuilder.setDeletable(true);
                        }
                        if (sink.containsKey(DorisSinkOptions.SINK_LABEL_PREFIX.key())) {
                            executionBuilder.setLabelPrefix(
                                    sink.get(DorisSinkOptions.SINK_LABEL_PREFIX.key())
                                            + "-"
                                            + getSinkSchemaName(table)
                                            + "_"
                                            + getSinkTableName(table));
                        } else {
                            executionBuilder.setLabelPrefix(
                                    "ztorn-"
                                            + getSinkSchemaName(table)
                                            + "_"
                                            + getSinkTableName(table)
                                            + UUID.randomUUID());
                        }
                        if (sink.containsKey(DorisSinkOptions.SINK_MAX_RETRIES.key())) {
                            executionBuilder.setMaxRetries(
                                    Integer.valueOf(
                                            sink.get(DorisSinkOptions.SINK_MAX_RETRIES.key())));
                        }

                        executionBuilder.setStreamLoadProp(properties).setDeletable(true);

                        DorisSink.Builder<String> builder = DorisSink.builder();
                        builder.setDorisReadOptions(DorisReadOptions.builder().build())
                                .setDorisExecutionOptions(executionBuilder.build())
                                .setDorisOptions(dorisOptions)
                                .setSerializer(
                                        JsonDebeziumSchemaSerializer.builder()
                                                .setDorisOptions(dorisOptions)
                                                .build());

                        process.getSideOutput(v)
                                .rebalance()
                                .sinkTo(builder.build())
                                .name(
                                        "Doris Schema Evolution Sink(table=["
                                                + getSinkSchemaName(table)
                                                + "."
                                                + getSinkTableName(table)
                                                + "])");
                    });
        }
        return dataStreamSource;
    }

    protected void buildColumn(
            List<String> columnNameList, List<LogicalType> columnTypeList, List<Column> columns) {
        for (Column column : columns) {
            columnNameList.add(column.getName());
            columnTypeList.add(getLogicalType(column));
        }
    }

    public LogicalType getLogicalType(Column column) {
        switch (column.getJavaType()) {
            case STRING:
                return new VarCharType();
            case BOOLEAN:
            case JAVA_LANG_BOOLEAN:
                return new BooleanType();
            case BYTE:
            case JAVA_LANG_BYTE:
                return new TinyIntType();
            case SHORT:
            case JAVA_LANG_SHORT:
                return new SmallIntType();
            case LONG:
            case JAVA_LANG_LONG:
                return new BigIntType();
            case FLOAT:
            case JAVA_LANG_FLOAT:
                return new FloatType();
            case DOUBLE:
            case JAVA_LANG_DOUBLE:
                return new DoubleType();
            case DECIMAL:
                if (column.getPrecision() == null || column.getPrecision() == 0) {
                    return new DecimalType(38, column.getScale());
                } else {
                    return new DecimalType(column.getPrecision(), column.getScale());
                }
            case INT:
            case INTEGER:
                return new IntType();
            case DATE:
            case LOCAL_DATE:
                return new DateType();
            case LOCAL_DATETIME:
            case TIMESTAMP:
                return new TimestampType();
            case BYTES:
                return new VarBinaryType(Integer.MAX_VALUE);
            default:
                return new VarCharType();
        }
    }

    protected Object convertValue(Object value, LogicalType logicalType) {
        if (value == null) {
            return null;
        }
        if (logicalType instanceof VarCharType) {
            return StringData.fromString((String) value);
        } else if (logicalType instanceof DateType) {
            return StringData.fromString(
                    Instant.ofEpochMilli((long) value)
                            .atZone(ZoneId.systemDefault())
                            .toLocalDate()
                            .toString());
        } else if (logicalType instanceof TimestampType) {
            return TimestampData.fromTimestamp(Timestamp.from(Instant.ofEpochMilli((long) value)));
        } else if (logicalType instanceof DecimalType) {
            final DecimalType decimalType = ((DecimalType) logicalType);
            final int precision = decimalType.getPrecision();
            final int scale = decimalType.getScale();
            return DecimalData.fromBigDecimal(
                    new BigDecimal(String.valueOf(value)), precision, scale);
        } else {
            return value;
        }
    }

    @Override
    public String getSinkSchemaName(Table table) {
        String schemaName = table.getSchema();

        if (config.getSink().containsKey("sink.db")) {
//            schemaName = config.getSink().get("sink.db");
            schemaName = SqlUtil.replaceAllParam(config.getSink().get(FlinkCDCConfig.SINK_DB), "schemaName", schemaName);
        }
        return schemaName;
    }

    @Override
    public String getSinkTableName(Table table) {
        String tableName = table.getName();
        if (config.getSink().containsKey("table.prefix.schema")) {
            if (Boolean.valueOf(config.getSink().get("table.prefix.schema"))) {
                tableName = table.getSchema() + "_" + tableName;
            }
        }
        if (config.getSink().containsKey("table.prefix")) {
            tableName = config.getSink().get("table.prefix") + tableName;
        }
        if (config.getSink().containsKey("table.suffix")) {
            tableName = tableName + config.getSink().get("table.suffix");
        }
        if (config.getSink().containsKey("table.lower")) {
            if (Boolean.valueOf(config.getSink().get("table.lower"))) {
                tableName = tableName.toLowerCase();
            }
        }
        if (config.getSink().containsKey("table.upper")) {
            if (Boolean.valueOf(config.getSink().get("table.upper"))) {
                tableName = tableName.toUpperCase();
            }
        }
        return tableName;
    }

    protected List<String> getPKList(Table table) {
        List<String> pks = new ArrayList<>();
        if (Asserts.isNullCollection(table.getColumns())) {
            return pks;
        }
        for (Column column : table.getColumns()) {
            if (column.isKeyFlag()) {
                pks.add(column.getName());
            }
        }
        return pks;
    }
}
